


		《 Linux_28_进程池 》


命令行中的 | ,就是匿名管道
[a@192 pipe]$ . sleep 100 | sleep 100 | sleep 100

进程池: 预先创建一批子进程
---------------------------------------------
           w    管道    r   worker/slave
         ┌─→ [       ]  ─→ 【 子进程 】
         │
 master  │─→ [       ]  ─→ 【 子进程 】
【父进程】{
         │─→ [       ]  ─→ 【 子进程 】
         │
         └─→ [       ]  ─→ 【 子进程 】
---------------------------------------------

【父进程】要进行后端任务划分的负载均衡!
[管道] 里面没有数据，[worker进程] 就在阻塞等待等待任务的到来!!!
master向哪个 [管道] 写入, 就是唤醒哪个【子进程】来处理任务


//——————————————————————————————————————————————


//——————————————————————————————————————————————


//      《 ProcessPool.cc 》


#define _CRT_SECURE_NO_WARNINGS 1
include <iostream>
#include <string>
#include <vector>
#include <unistd.h>
#include <sys/types.h>


void work(int rfd)
{
    while (true) sleep(1);   //让子进程做的事
}

//master 
class Channel
{
public:
    Channel(int wfd, pid_t id, const std::string& name)
        :_wfd(wfd), _subprocessid(id), _name(name)
    { }
    int GetWfd() { return _wfd; }                   //获得写入文件描述符
    pid_t GetProcessId() { return _subprocessid; }  //子进程pid
    std::string GetName() { return _name; }           //名字
    ~Channel()
    {

    }

private:
    int _wfd;                //写文件描述符
    pid_t _subprocessid;     //子进程ID
    std::string _name;       //给信道起名字
};

// ./processpool 5  例如想要创建5个进程  
int main(int argc, char* argv[])     //输入的命令行参数 processpool 5 如果 ！= 2 个。
{
    if (argc != 2)           //如果不够
    {
        std::cerr << "Usage: " << argv[0] << "processnum" << std::endl;
        return 1;
    }
    // （int argc, char* argv[]）==> argv[0], argv[1] 是 main()的两个参数
    //  num 就是我们命令行输入的第2个参数，要创建的 子进程 个数
    int num = std::stoi(argv[1]);   //argv[1]是字符串类型，需要字符串转整数：stoi 

    std::vector<Channel> channels;
    //创建信道和子进程
    for (int i = 0; i < num; i++)    //通过循环的方式来创建一大堆进程
    {
        //1.创建管道
        int pipefd[2] = { 0 };
        int n = pipe(pipefd);
        if (n < 0) exit(1);          //管道创建失败，直接终止。

        //2.创建子进程   
        pid_t id = fork();
        if (id == 0)                 //如果 fork 成功了，会返回子进程的 PID 给父进程 and 0
        {
            close(pipefd[1]);       //子进程关闭 写
            work(pipefd[0]);
            close(pipefd[0]);
            exit(0);
        }

        //3.构建Channel名字
        std::string channel_name = "Channel_" + std::to_string(i);

        //父进程 关闭读，(让父进程写，子进程读)
        close(pipefd[0]);
        //a.子进程的pid,    b.父进程关系的管道的w端
        channels.push_back(Channel(pipefd[1], id, channel_name));
    }

    //测试：
    for (auto& channel : channels)
    {
        std::cout << "====================================" << std::endl;
        std::cout << channel.GetName() << std::endl;
        std::cout << channel.GetWfd() << std::endl;
        std::cout << channel.GetProcessId() << std::endl;
    }

    sleep(100);
    return 0;
}

——————————————————————————————————————————————


——————————————————————————————————————————————

创建5个进程
[a@192 ProcessPool]$ ./processpool 5
====================================
Channel_0
4
12105
====================================
Channel_1
5
12106
====================================
Channel_2
6
12107
====================================
Channel_3
7
12108
====================================
Channel_4
8
12109

-i 忽略大小写
while :; do ps axj | head -1 && ps axj | grep -i ProcessPool | grep -v grep; echo"--------------------"； sleep 1; done
  PPID    PID   PGID    SID TTY       TPGID STAT   UID   TIME COMMAND
bash: echo--------------------；: command not found...
  PPID    PID   PGID    SID TTY       TPGID STAT   UID   TIME COMMAND
bash: echo--------------------；: command not found...


——————————————————————————————————————————————


——————————————————————————————————————————————


    《 优化 》

#define _CRT_SECURE_NO_WARNINGS 1
include <iostream>
#include <string>
#include <vector>
#include <unistd.h>
#include <sys/types.h>


void work(int rfd)
{
    while (true) sleep(1);   //让子进程做的事
}

//master 
class Channel
{
public:
        //（文件描述符， 文件描述符ID，  对应的名字      ）
    Channel(int wfd, pid_t id, const std::string& name)
        :_wfd(wfd), _subprocessid(id), _name(name)  //初始化
    { }
    int GetWfd() { return _wfd; }                   //获得写入文件描述符
    pid_t GetProcessId() { return _subprocessid; }  //子进程pid
    std::string GetName() { return _name; }           //名字
    ~Channel()
    {

    }

private:
    int _wfd;
    pid_t _subprocessid;
    std::string _name;      //给信道起名字
};

// 形参类型和命名规范：
// const &  输入型参数
// &        输入输出型参数
// *        输出型参数

//创建信道和子进程
void CreateChannelAndSub(int num, std::vector<Channel> *channels)
{
    //BUG?
   //创建信道和子进程
    for (int i = 0; i < num; i++)
    {
        //1.创建管道
        int pipefd[2] = { 0 };
        int n = pipe(pipefd);
        if (n < 0) exit(1);          //管道创建失败，直接终止。

        //2.创建子进程   
        pid_t id = fork();
        if (id == 0)
        {
            close(pipefd[1]);       //子进程关闭 写
            work(pipefd[0]);
            close(pipefd[0]);
            exit(0);
        }

        //3.构建Channel名字
        std::string channel_name = "Channel_" + std::to_string(i);

        //父进程 关闭读，(让父进程写，子进程读)
        close(pipefd[0]);
        //a.子进程的pid,    b.父进程关系的管道的w端
        channels->push_back(Channel(pipefd[1], id, channel_name));
    }

}

// ./processpool 5  例如想要创建5个进程
int main(int argc, char* argv[])
{
    if (argc != 2)
    {
        std::cerr << "Usage: " << argv[0] << "processnum" << std::endl;
        return 1;
    }

    int num = std::stoi(argv[1]);   //字符串转整数：stoi 

    std::vector<Channel> channels;
    //1.创建信道和子进程
    CreateChannelAndSub(num, &channels); //创建 num 个子进程， 放在&channels 中

    //测试：
    for (auto& channel : channels)
    {
        std::cout << "====================================" << std::endl;
        std::cout << channel.GetName() << std::endl;
        std::cout << channel.GetWfd() << std::endl;
        std::cout << channel.GetProcessId() << std::endl;
    }

    sleep(100);
    return 0;
}

——————————————————————————————————————————————

测试结果：
[a@192 ProcessPool]$ ./processpool 4
====================================
Channel_0
4
7675
====================================
Channel_1
5
7676
====================================
Channel_2
6
7677
====================================
Channel_3
7
7678
  ... ...

//——————————————————————————————————————————————

[a@192 ~]$ while :; do ps axj | head -1 && ps axj | grep -i ProcessPool | grep -v grep; echo"--------------------"； sleep 1; done
  PPID    PID   PGID    SID TTY       TPGID STAT   UID   TIME COMMAND
  2569   7673   7673   2569 pts/0      7673 S+    1000   0:00 ./processpool 4
  7673   7675   7673   2569 pts/0      7673 S+    1000   0:00 ./processpool 4
  7673   7676   7673   2569 pts/0      7673 S+    1000   0:00 ./processpool 4
  7673   7677   7673   2569 pts/0      7673 S+    1000   0:00 ./processpool 4
  7673   7678   7673   2569 pts/0      7673 S+    1000   0:00 ./processpool 4
bash: echo--------------------；: command not found...
  ... ...


——————————————————————————————————————————————


——————————————————————————————————————————————



		《 Task.hpp 》


#pragma once

#include <iostream>
#include <ctime>
#include <cstdlib>
#include <unistd.h>
#include <sys/types.h>

#define TaskNum 3


typedef void (*task_t) ();  //task_t 函数指针类型

void Print()    //打印
{
    std::cout << "I am print task" << std::endl;
}
void DownLoad() //下载
{
    std::cout << "I am a download task" << std::endl;
}
void Flush()    //刷新
{
    std::cout << "I am a flush task" << std::endl;
}


task_t tasks[TaskNum];

//加载任务
void LoadTask()
{
    //产生随机数
    srand(time(nullptr) ^ getpid() ^ 17777);    // ^ 增加随机性
    tasks[0] = Print;
    tasks[1] = DownLoad;
    tasks[2] = Flush;
}

//执行任务
void ExcuteTask(int number)
{
    if(number < 0 || number > TaskNum - 1) return;  //如果任务不在我们的任务码范围直接返回。
    tasks[number] (); // void (*task_t) (); //task_t 函数指针类型 ==> task_t tasks[TaskNum]; 通过数组下标再回去调用对应的任务函数
}

//随机选择任务
int SelectTask()
{
    return rand() % TaskNum;
}


//		《 ProcessPool.cc 》

#define _CRT_SECURE_NO_WARNINGS 1
#include <iostream>
#include <string>
#include <vector>
#include <unistd.h>
#include <sys/types.h>
#include "Task.hpp"

void work(int rfd)
{   //让子进程做的事
    while (true)
    {
        int command = 0;
        // 读到的内容是 父进程写进来的，写了一个整数，就读到一个整数。
        int n = read(rfd, &command, sizeof(command));   //rfd == pipefd[0] 文件描述符：读端
        if(n == sizeof(int))    // 读 写 的数量要相等
        {
            std::cout << "pid is : " << getpid() << " handler task" << std::endl;
            ExcuteTask(command);    //执行任务
        }
    }
}

//master 
class Channel
{
public:
    Channel(int wfd, pid_t id, const std::string& name)
        :_wfd(wfd), _subprocessid(id), _name(name)
    { } 
    int GetWfd() { return _wfd; }                   //获得写入文件描述符
    pid_t GetProcessId() { return _subprocessid; }  //子进程pid
    std::string GetName() { return _name; }           //名字
    ~Channel()
    {

    }

private:
    int _wfd;               //写文件描述符
    pid_t _subprocessid;    //子进程ID
    std::string _name;      //给信道起名字
};

//创建信道和子进程
void CreateChannelAndSub(int num, std::vector<Channel> *channels)
{
    //BUG?
   //创建信道和子进程 
    for (int i = 0; i < num; i++)
    {
        //1.创建管道
        int pipefd[2] = { 0 };
        int n = pipe(pipefd);
        if (n < 0) exit(1);          //管道创建失败，直接终止。

        //2.创建子进程   
        pid_t id = fork();
        if (id == 0)
        {
            close(pipefd[1]);       //子进程关闭 写
            work(pipefd[0]);
            close(pipefd[0]);
            exit(0);
        }

        //3.构建Channel名字
        std::string channel_name = "Channel_" + std::to_string(i);

        //父进程 关闭读，(让父进程写，子进程读)
        close(pipefd[0]);
        //a.子进程的pid,    b.父进程关系的管道的w端
        channels->push_back(Channel(pipefd[1], id, channel_name));
    }

}

//b.选择一个信道和进程
int NextChannel(int channelnum)
{
    static int next = 0;
    int channel = next;
    next++; 
    next %= channelnum;     //控制在信道数量范围内，从0 ~ channelnum 循环
    return channel;
}
  
//c.发送任务                   
void SendTaskCommand(Channel&channel, int taskcommand)
{
    //（向指定的文件描述符中，发taskcommand， 发多少字节 ）
    write(channel.GetWfd(), &taskcommand, sizeof(taskcommand));
}

//./processpool 5  例如想要创建5个进程
int main(int argc, char* argv[])
{
    if (argc != 2)
    {
        std::cerr << "Usage: " << argv[0] << "processnum" << std::endl;
        return 1;
    }

    int num = std::stoi(argv[1]);   //字符串转整数：stoi 
    LoadTask();                     //加载任务

    std::vector<Channel> channels;
    //1.创建信道和子进程
    CreateChannelAndSub(num, &channels);

    //2.通过Channel控制子进程
    while(true)
    {
        sleep(1);
        //a.(随机)选择一个任务
        int taskcommand = SelectTask();
        //b.选择一个信道和进程
        int channel_index = NextChannel(channels.size());
        //c.发送任务
        SendTaskCommand(channels[channel_index], taskcommand);  //向指定的(随机)信道中，发送指定(0 ~ channelnum)的任务码。

        std::cout << "----------------------------------" << std::endl;
        std::cout << "taskcommand: " << taskcommand << " channel: "\
        << channels[channel_index].GetName() << " sub process: "\
        << channels[channel_index].GetProcessId() << std::endl;
    }

    //3.回收管道和子进程

    return 0;
}

——————————————————————————————————————————————

创建了3 个子进程，让它们轮回执行随机任务：
[a@192 ProcessPool]$ ./processpool 3
----------------------------------
taskcommand: 2 channel: Channel_0 sub process: 【102794】
pid is : 【102794】 handler task
I am a flush task
----------------------------------
taskcommand: 2 channel: Channel_1 sub process: {102795}
pid is : {102795} handler task
I am a flush task
----------------------------------
taskcommand: 0 channel: Channel_2 sub process: [102796]
pid is : [102796] handler task
I am print task
----------------------------------
taskcommand: 0 channel: Channel_0 sub process: 【102794】
pid is : 【102794】 handler task
I am print task
----------------------------------
taskcommand: 2 channel: Channel_1 sub process: {102795}
pid is : {102795} handler task
I am a flush task
----------------------------------
taskcommand: 0 channel: Channel_2 sub process: [102796]
pid is : [102796] handler task
I am print task
----------------------------------

窗口2：
[a@192 ~]$ while :; do ps axj | head -1 && ps axj | grep -i ProcessPool | grep -v grep; echo"--------------------"； sleep 1; done
  2569 102793 102793   2569 pts/0    102793 S+    1000   0:00 ./processpool 3
102793 【102794】 102793   2569 pts/0    102793 S+    1000   0:00 ./processpool 3
102793  {102795} 102793   2569 pts/0    102793 S+    1000   0:00 ./processpool 3
102793  [102796] 102793   2569 pts/0    102793 S+    1000   0:00 ./processpool 3
bash: echo--------------------；: command not found...

——————————————————————————————————————————————